package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.common.constant.MqttDestinationConst;
import cn.tedu.charging.common.pojo.entity.ProgressEntity;
import cn.tedu.charging.common.pojo.message.ChargingCheckResultMessage;
import cn.tedu.charging.common.pojo.message.ChargingProgressMsg;
import cn.tedu.charging.common.pojo.message.ProgressData;
import cn.tedu.charging.common.pojo.param.ProgressCostParam;
import cn.tedu.charging.common.pojo.po.ChargingBillFailPO;
import cn.tedu.charging.common.pojo.po.ChargingBillSuccessPO;
import cn.tedu.charging.common.pojo.vo.ProgressCostVO;
import cn.tedu.charging.common.protocol.WebSocketMessage;
import cn.tedu.charging.common.utils.SnowflakeIdGenerator;
import cn.tedu.charging.order.feign.DeviceClient;
import cn.tedu.charging.order.mapper.BillFailMapper;
import cn.tedu.charging.order.mapper.BillSuccessMapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.BeanUtils;

import java.nio.charset.StandardCharsets;
import java.util.Date;

/**
 * 这是一个要交给mqttClient连接对象使用的回调对象
 * 包含一个底层连接和emqx通信的4个回调时间点
 */
@Slf4j
public class MqttMyCallback implements MqttCallbackExtended {
    private MqttClient mqttClient;
    private MqttContext mqttContext;
    private SnowflakeIdGenerator snowflakeIdGenerator=new SnowflakeIdGenerator(1,1);
    public MqttMyCallback(
            MqttClient mqttClient,MqttContext mqttContext){
        this.mqttClient=mqttClient;
        this.mqttContext=mqttContext;
    }
    /**
     * 回调时机: 客户端mqttClient成功建立与emqx服务端连接关系时调用
     * boolean reconnect: true表示本次连接是重连,false表示第一次连接
     * String serverUri: 实际连接的地址
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        if(reconnect){
            log.info("自动重连成功,borker服务端地址:{}",serverURI);
        }else{
            log.info("首次连接成功,borker服务端地址:{}",serverURI);
        }
        //利用当前回调类所属的客户端,进行连接成功之后的主题订阅 如果当前回调类所属客户端没有订阅任何主题
        //不可能消费主题的消息
        try{
            //共享订阅 设备反馈自检结果
            mqttClient.subscribe(
                    "$share/order/"+ MqttDestinationConst.GUN_CHECK_RESULT_TOPIC,
                    1);
            //共享订阅 订单订阅设备充电进度
            mqttClient.subscribe("$share/order/"+MqttDestinationConst.CHARGING_PROGRESS_TOPIC,
                    1);
        }catch (Exception e){
            log.error("订阅主题失败",e);
        }
    }

    /**
     * 回调时机: mqttClient连接断开时调用
     * Throwable cause: 连接断开原因
     * 如果设置了setAutomaticReconnect(true).自动重连,否则这个代码 手动处理重新连接
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.info("连接断开,原因:{}",cause.getMessage());
    }

    /**
     * 回调时机: mqttClient通过订阅主题 获取到消息的时候 所以这个方法中应该包含消费逻辑
     * String topic: 消息来源主题
     * MqttMessage message: 消息对象(payLoad,qos retain)
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.info("收到消息,来源topic:{},消息内容:{}",topic,
                new String(message.getPayload(), StandardCharsets.UTF_8));
        //判断解析消息 设备反馈消息
        if (topic!=null&&topic.equals(MqttDestinationConst.GUN_CHECK_RESULT_TOPIC)){
            //对消息做反序列化
            MySerializer<ChargingCheckResultMessage> mySerializer = mqttContext.getMySerializer();
            //获取消息体反序列化
            ChargingCheckResultMessage checkResult = mySerializer.deserialize(message.getPayload(), ChargingCheckResultMessage.class);
            //调用业务方法处理
            //TODO redis客户端 lockKey value doWhile抢锁
            handleCheckResult(checkResult);
            //TODO 释放锁
        }else if (topic!=null&&topic.equals(MqttDestinationConst.CHARGING_PROGRESS_TOPIC)){
            MySerializer<ChargingProgressMsg> mySerializer = mqttContext.getMySerializer();
            ChargingProgressMsg progressMsg = mySerializer.deserialize(message.getPayload(), ChargingProgressMsg.class);
            handleProgressMsg(progressMsg);
        }
    }

    private void handleProgressMsg(ChargingProgressMsg progressMsg) {
        //1.远程调用cost服务,获取本次充电进度计价结果 得到总价和总时间
        ProgressCostParam param = new ProgressCostParam();
        //消息属性和 接口入参属性一一对应
        BeanUtils.copyProperties(progressMsg,param);
        ProgressCostVO costVO=mqttContext.getCostClient().calculateCost(param).getData();
        //2.将充电进度存储到es
        ProgressEntity entity=new ProgressEntity();
        BeanUtils.copyProperties(progressMsg,entity);
        //总金额 补充
        entity.setTotalCost(costVO.getTotalCost());
        //id补充 雪花算法 保证订单充电进度 时序正常的
        entity.setId(snowflakeIdGenerator.nextId()+"");
        mqttContext.getProgressEsRepository().save(entity);
        //3.根据充电进度 充电是否满了,判断给用户推送什么消息
        if (progressMsg.getIsFull()){
            //按照弹窗处理
            WebSocketMessage<String> message=new WebSocketMessage<>();
            message.setState(1);//失败异常
            message.setMessage("充满");
            message.setData("A已充满,请结束订单,支付");
            mqttContext.getWebSocketServer().sendMsgToUser(progressMsg.getUserId(),message);
        }else{
            //假设 当前充电度数 16度 计算总充电量100 16%
            //当前业务中  有专门提示用户充电进度的实时数据封装
            WebSocketMessage<ProgressData> message=new WebSocketMessage<>();
            message.setState(3);//state=3表示充电进度展示
            message.setMessage("充电中");
            ProgressData data=new ProgressData();
            data.setTotalCapacity(100.00);//写死的充满电度数
            data.setTotalCost(costVO.getTotalCost());
            String totalCapacityKey="charging:total:capacity:"+progressMsg.getOrderNo();
            Double chargingCapacity = progressMsg.getChargingCapacity();//一次充电度数
            Double totalCap = mqttContext.getRedisTemplate().opsForValue().increment(totalCapacityKey, chargingCapacity);
            data.setChargingCapacity(totalCap);
            data.setHours(0);//可以利用totalTime api调用转化成小时数
            data.setMinutes(10);
            data.setSeconds(10);
            message.setData(data);
            mqttContext.getWebSocketServer().sendMsgToUser(progressMsg.getUserId(),message);
        }
    }

    private void handleCheckResult(ChargingCheckResultMessage checkResult) {
        //消费者拿到设备反馈 最核心的功能 根据反馈true和false新增success和fail
        //先判断 有没有 在新增
        //1.利用订单编号orderNo  billId 查询存在订单 success|fail
        String billId = checkResult.getOrderNo();
        //利用count函数检查存在或者不存在 select count(*) form success where bill_id=?
        QueryWrapper queryWrapper=new QueryWrapper();
        queryWrapper.eq("bill_id",billId);
        Long exSuccess=mqttContext.getBillSuccessMapper().selectCount(queryWrapper);
        Long exFail=mqttContext.getBillFailMapper().selectCount(queryWrapper);
        //2.判断是否存在 存在则结束 不存在下一步 只要上述两个count有一个>0 则结束
        if(exSuccess>0l||exFail>0l){
            log.warn("订单:{}重复消费",billId);
            return;
        }
        //3.判断true|false
        Boolean result = checkResult.getResult();
        if (result){
            //3.1 true 组织successPO写入success表格 发送命令开始充电(undo)
            log.debug("订单:{},设备自检结果成功",billId,result);
            ChargingBillSuccessPO po=new ChargingBillSuccessPO();
            //订单编号 user用户身份 创建时间 修改时间 订单刚创建初始状态 1=订单正在充电中
            po.setBillStatus(1);
            po.setBillId(billId);
            po.setUserId(checkResult.getUserId());
            po.setCreateTime(new Date());
            po.setUpdateTime(new Date());
            //缺少的字段 考虑重新定义接口 让设备传递的消息 包含更多属性
            mqttContext.getBillSuccessMapper().insert( po);
            //UNDO 设备自检成功 订单创建成功后 给设备发送第二个命令 开始给车充电
        }else{
            log.debug("订单:{},设备自检结果失败",billId,result);
            //3.2 false 组织failPO写入fail表格,调用设备接口修改枪状态(undo) 实时消息推送
            ChargingBillFailPO po=new ChargingBillFailPO();
            //订单编号 用户id 时间
            po.setBillId(billId);
            po.setUserId(checkResult.getUserId());
            po.setCreateTime(new Date());
            po.setUpdateTime(new Date());
            po.setFailDesc("设备自检失败");
            mqttContext.getBillFailMapper().insert(po);
            //设计一个修改枪状态的device服务接口
            mqttContext.getDeviceClient().updateGunStatus(checkResult.getGunId(),3);
            //实时消息推送 告知设备故障 或者其他原因 订单无法继续充电 请换枪 赠送优惠券
            //订单失败 组织webSocketMessage 给用户userId推送消息
            WebSocketMessage webSocketMessage=new WebSocketMessage();
            //state=1表示订单有问题 2用户余额不足 3正常同步充电进度
            webSocketMessage.setState(1);
            webSocketMessage.setMessage("订单有问题");
            webSocketMessage.setData("B扫码下单失败,请换枪,非常抱歉,赠送您一张优惠券http://AN0df32fj3");
            mqttContext.getWebSocketServer().sendMsgToUser(checkResult.getUserId(),webSocketMessage);
        }
    }

    /**
     * 回调时机: mqttClient投递发送消息的时候 接收到服务端反馈确认成功的 消息发送成功调用
     * 但是只有在QOS>0的时候才有效 因为  QOS=0 客户端不关心成功失败.
     * IMqttDeliveryToken token: 包含消息ID等上下文信息 原始消息就在token里
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        //可以从token获得发送的原始消息
        String body=null;
        try{
            MqttMessage message = token.getMessage();
            body=new String(message.getPayload(),StandardCharsets.UTF_8);
        }catch (Exception e){
            log.error("获取原始消息失败",e);
        }
        log.info(
                "消息投递完成,消息ID:{},消息原始内容:{}",
                token.getMessageId(),body
                );
    }
}
